package com.atguigu.ct.analysis.io;

import com.atguigu.ct.analysis.kv.AnalysisKey;
import com.atguigu.ct.analysis.kv.AnalysisValue;
import com.atguigu.ct.common.util.JDBCUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;

/**
 * Description: MySQL数据格式化输入对象
 * @Author: yyx
 * @Create: 2023/5/5 19:52
 * @Version: 8
 */
public class MySQLBeanOutPutFormat extends OutputFormat<AnalysisKey, AnalysisValue> {
    //getRecordWriter的静态内部类
    protected static class MySQLRecordWriter extends RecordWriter<AnalysisKey,AnalysisValue>{
        private Connection connection = null;
        Map<String,Integer> userMap = new HashMap<String,Integer>();//时间和用户的映射，避免冗余
        Map<String,Integer> dateMap = new HashMap<String,Integer>();//时间和日期的映射，避免冗余

        /**
         * 获取资源
         */
        public MySQLRecordWriter() {
            connection = JDBCUtil.getConnection();
            PreparedStatement pstat = null;
            ResultSet rs = null;
            try {
                //读取用户数据
                String queryUserSql = "select id,tel from ct_user";
                pstat = connection.prepareStatement(queryUserSql);
                rs = pstat.executeQuery();
                while (rs.next()){
                    Integer id = rs.getInt(1);
                    String tel = rs.getString(2);
                    userMap.put(tel,id);
                }
                rs.close();
                //读取时间数据
                String queryDateSql = "select id,year,month,day from ct_date";
                pstat = connection.prepareStatement(queryDateSql);
                rs = pstat.executeQuery();
                while (rs.next()){
                    Integer id = rs.getInt(1);
                    String year = rs.getString(2);
                    String month = rs.getString(3);
                    String day = rs.getString(4);
                    if(month.length() == 1){
                        month = "0" + month;
                    }
                    if(day.length() == 1){
                        day = "0" + day;
                    }
                    dateMap.put(year+month+day,id);
                }
                rs.close();
            } catch (SQLException throwables) {
                throwables.printStackTrace();
            } finally {
                if (rs != null) {
                    try {
                        rs.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
                if (pstat != null) {
                    try {
                        pstat.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        /**
         * 输出数据
         * @param key
         * @param value
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        public void write(AnalysisKey key, AnalysisValue value) throws IOException,
                InterruptedException {
            PreparedStatement pstat = null;
            try {
                String insertSQL = "insert into ct_call(telid,dateid,sumcall,sumduration) values(?,?,?,?)";
                pstat = connection.prepareStatement(insertSQL);
                pstat.setInt(1,userMap.get(key.getTel()));
                pstat.setInt(2,dateMap.get(key.getDate()));
                pstat.setInt(3,Integer.parseInt(value.getSumCall()));//转为整形
                pstat.setInt(4,Integer.parseInt(value.getSumDuration()));
                pstat.executeUpdate();
            } catch (SQLException e) {
                e.printStackTrace();
            } finally {
                if(pstat != null){
                    try {
                        pstat.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        /**
         * 释放资源
         * @param taskAttemptContext
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
            if(connection != null){
                try {
                    connection.close();
                } catch (SQLException throwables) {
                    throwables.printStackTrace();
                }
            }
        }
    }
    @Override
    public RecordWriter<AnalysisKey, AnalysisValue> getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        return new MySQLRecordWriter();
    }

    @Override
    public void checkOutputSpecs(JobContext jobContext) throws IOException, InterruptedException {
    }
    private FileOutputCommitter committer = null;
    public static Path getOutputPath(JobContext job){
        String name = job.getConfiguration().get(FileOutputFormat.OUTDIR);
        return name == null ? null : new Path(name);
    }
    @Override
    public OutputCommitter getOutputCommitter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        if(committer == null){
            Path output = getOutputPath(taskAttemptContext);
            committer = new FileOutputCommitter(output,taskAttemptContext);
        }
        return committer;
    }
}
